package chapter04

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions.{col, desc, sum, trim}

object Test10_restaurant {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME","root")
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("restaurant")
      .getOrCreate()
    import spark.implicits._
    //读文件 csv
    val df = spark.read
      .option("inferSchema", "true")
      .option("header", "true")
      .csv("input/restaurant.csv")
    df.printSchema()
    //计算口味大于7分的数据
    //df.filter("口味>7").show 如果列名为中文 使用条件sql不成立 只能先把列名换成英文
    df.withColumnRenamed("口味","taste")
      .filter("taste>7").show(5)
    //只能使用col
    df.filter(col("口味")>7).show(5)
    //删除指定列
    val df1 = df.drop("_c10", "_c11")
    df1.show(5)
    //删除缺失数据的字段
    println(df1.count())
    val df2 = df1.na.drop()
    println(df2.count())
    //统计各类别餐饮店点评数，并按降序排列
    df2.groupBy("类别")
      .sum("点评数")
      .orderBy(desc("sum(点评数)"))
      .limit(10)
      .repartition(1)
      .show(5)
    //各个区分数大于7饭店的个数，按照降序排序
    df2.filter($"口味">7)
      .groupBy("行政区")
      .count()
      .orderBy($"count".desc)
      .show(10)
    //每个区的评分的平均数 评分=(口味+环境+服务)/3
    df2.withColumn("评分",($"口味"+$"环境"+$"服务")/3)
      .groupBy("行政区")
      .avg("评分")
      .sort($"avg(评分)".desc)
      .show(10)
    //每个区的人均消费数最高的类别
    df2.groupBy("行政区","类别")
      .max("人均消费")
      .repartition(1)
      .write
      .option("header","true")
      .partitionBy("行政区")
      .mode(SaveMode.Overwrite)
      .csv("input/taste")
    //每个类别的人均消费数
    df2.filter($"人均消费"=!=0)
      .groupBy("类别")
      .avg("人均消费")
      .sort($"avg(人均消费)".desc)
      .show()
    //所有浦东新区的数据
    df2.filter($"行政区"===" 浦东新区").show()
    //过滤空格
    df2.withColumn("行政区",trim($"行政区"))
      .filter($"行政区"==="浦东新区").show()
    //定义函数
    val res = spark.udf.register("res", (value: String) => {
      value.trim
    })
    df2.withColumn("行政区",res($"行政区"))
      .filter($"行政区"==="浦东新区").show()
    val test2 = spark.udf.register("test", (value1: Double, value2: String) => {
      value1 + value2
    })
    df2.withColumn("newTest",test2($"口味",$"行政区")).show(10)
    println(df2.col("口味"))
  }
}
